/**
 * Copyright 2009 Takahiko Ito
 *
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.unigram.oluolu.ngram;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.conf.Configuration;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;

import org.unigram.oluolu.common.OluoluConstants;
import org.unigram.oluolu.common.OluoluException;
import org.unigram.oluolu.common.Query;
import org.unigram.oluolu.extractor.IExtractor;

/**
 *
 */
final class StringUtils {
    
    /** for safe. */
    private StringUtils() {}

    /**
     * Create one string connected with a separator from 
     * string array.
     * 
     * @param arry string array.
     * @param with separator
     * @return joined string with separator
     */
    public static String join(final String[] arry,
            final String with) {
        StringBuffer buf = new StringBuffer();
        for (String s : arry) {
            if (buf.length() > 0) {
                buf.append(with);
            }
            buf.append(s);
        }
        return buf.toString();
    }
}

/**
 *
 */
public class FrequentNGramMapper extends
        Mapper<LongWritable, Text, Text, IntWritable> {
    
    /** minimum ngram. */
    private int minimumGram;
    
    /** maximum ngram. */
    private int maximumGram;
    
    /** extractor which extracts Query objects from input. */
    private IExtractor extractor;

    /**
     * Map method.
     * @param key - dummy
     * @param value - line containing query log information
     * @param context - flush query and the frequencies
     * @throws IOException -
     * @throws InterruptedException -
     */
    @Override
    public final void map(final LongWritable key, final Text value,
            final Context context)
            throws IOException, InterruptedException {
        try {
            Query query = extractor.extract(value.toString());
            this.extractNGrams(query.getQueryString().toString(), context);
        } catch (OluoluException e) {
            e.printStackTrace();
        }
    }

    /**
     * Extract Ngrams from input String.
     *
     * @param inputStr query log
     * @param context flush query
     * @throws InterruptedException -
     * @throws IOException -
     */
    private void extractNGrams(final String inputStr,
            final Context context)
            throws IOException, InterruptedException {
        String[] queryAry = inputStr.split(" ");
        List<String> qlist = Arrays.asList(queryAry);
        for (int position = 0; position < qlist.size(); position++) {
            for (int gram = this.minimumGram; gram
                <= this.maximumGram; gram++) {
                if (position + gram <= qlist.size()) {
                    String keyStr = StringUtils.join(
                            (String[]) qlist.subList(position,
                                    (position + gram)).toArray(new String[0]),
                            " ");
                    context.write(new Text(keyStr), new IntWritable(1));
                }
            }
        }
    }

    /**
     * Setup.
     * 
     * @param context object to get needed system properties.
     */
    @Override
    public final void setup(final Context context) {
        Configuration jc = context.getConfiguration();
        this.minimumGram = NgramConstants.DEFAULT_NGRAM_MINIMUM_GRAM;
        this.maximumGram = NgramConstants.DEFAULT_NGRAM_MAXIMUM_GRAM;
        String extractorClassName = OluoluConstants.DEFAULT_EXTRACTOR;
        if (context == null || jc == null) {
            jc = new Configuration(); // added default configuration for testing
        }

        this.minimumGram = jc.getInt(NgramConstants.NGRAM_MINIMUM_GRAM,
                NgramConstants.DEFAULT_NGRAM_MINIMUM_GRAM);
        this.maximumGram = jc.getInt(NgramConstants.NGRAM_MAXIMUM_GRAM,
                NgramConstants.DEFAULT_NGRAM_MAXIMUM_GRAM);
        try {
            extractorClassName = jc.get(OluoluConstants.EXTRACTOR,
                    OluoluConstants.DEFAULT_EXTRACTOR);
            Class<? extends IExtractor> extractorClass = Class.forName(
                    extractorClassName).asSubclass(IExtractor.class);
            Constructor<? extends IExtractor> constructor = extractorClass
                    .getConstructor(Configuration.class);
            extractor = constructor.newInstance(jc);
        } catch (NoSuchMethodException nsme) {
            throw new RuntimeException(nsme);
        } catch (ClassNotFoundException cnfe) {
            throw new RuntimeException(cnfe);
        } catch (InstantiationException ie) {
            throw new RuntimeException(ie);
        } catch (IllegalAccessException iae) {
            throw new RuntimeException(iae);
        } catch (InvocationTargetException ite) {
            throw new RuntimeException(ite.getCause());
        }
    }
}